package com.fuchuang.seckillsystem.rocketmq;

import com.fuchuang.seckillsystem.dto.KeyConstants;
import com.fuchuang.seckillsystem.dto.TimeConstants;
import com.fuchuang.seckillsystem.exception.GlobalException;
import com.fuchuang.seckillsystem.vo.RespBeanEnum;
import com.fuchuang.seckillsystem.vo.SeckillMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class OrderProducer {
    
    private static final String TOPIC = "seckill";
    
    private static final String CREATE_TAG = "create";
    
    private static final String EXPIRE_TAG = "expire";
    
    private static final String USER = "user";
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    public void sendCreateOrderMessage(SeckillMessage message) {
        rocketMQTemplate.asyncSend(TOPIC + ":" + CREATE_TAG, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("创建订单信息成功：" + message);
            }
            
            @Override
            public void onException(Throwable throwable) {
                log.warn("创建订单信息失败：" + message);
                //消息发送失败，删除用户购买记录，回滚库存
                redisTemplate.boundHashOps(KeyConstants.SECKILL_ACTIVITY + message.getActivityId()).delete(KeyConstants.SECKILL_REPEAT + message.getGoodsId() + ":" + message.getUserId());
                redisTemplate.boundHashOps(KeyConstants.SECKILL_ACTIVITY + message.getActivityId()).increment(KeyConstants.SECKILL_GOODS_STOCK + message.getGoodsId(), 1);
                throw new GlobalException(RespBeanEnum.SECKILL_ERROR);
            }
        });
    }
    
    public void sendExpireOrderMessage(Integer orderId) {
        Message<Integer> build = MessageBuilder.withPayload(orderId).build();
        SendResult sendResult = rocketMQTemplate.syncSend(TOPIC + ":" + EXPIRE_TAG, build, 2000, TimeConstants.ORDER_EXPIRE_LEVEL);
        SendStatus sendStatus = sendResult.getSendStatus();
        //如果发生失败，直接抛异常，交给orderService处理
        if (sendStatus != SendStatus.SEND_OK) {
            throw new GlobalException(RespBeanEnum.SECKILL_ERROR);
        }
    }

//    public void sendRegisterUser(User user) {
//        rocketMQTemplate.asyncSend(TOPIC + ":" + USER, user, new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                System.out.println("发生成功");
//            }
//
//            @Override
//            public void onException(Throwable throwable) {
//                System.out.println("发生失败");
//            }
//        });
//    }
    
    
}
